using System;
using System.Collections.Generic;
using System.Threading;
using SS.Common.Logging;

namespace SS.Common.Threading.JobExecutor
{
    using SS.Common.Collection;

    public sealed class NonQueueJobExcutor<T> : IJobExecutor<T>
        where T : class, IJob
    {
        /// <summary>
        /// Indicate the executor is running
        /// </summary>
        private const int RUNNING = 0;

        /// <summary>
        /// Indicate the executor is shutting down
        /// </summary>
        private const int SHUTDOWN = 1;

        /// <summary>
        /// Indicate the executor is stop
        /// </summary>
        private const int STOP = 2;

        /// <summary>
        /// Indicate the executor is terminated
        /// </summary>
        private const int TERMINATED = 3;

        /// <summary>
        /// Minimum number of threads will be maintained
        /// </summary>
        private volatile int minThead;

        /// <summary>
        /// Maximum number of threads will be maintained
        /// </summary>
        private volatile int maxThead;

        /// <summary>
        /// How long to release idle thread in milliseconds
        /// </summary>
        private volatile int keepAliveTime;

        /// <summary>
        /// Job jobQueue
        /// </summary>
        private readonly WaitBlockingCollection<T> jobQueue;

        /// <summary> 
        /// Queue size check interval
        /// </summary> 
        private int intervalQueueSizeCheck = 0;

        private static readonly ILogger Logger = LoggerManager.GetLogger(typeof(JobExecutorEx<T>).Name);

        /// <summary>
        /// Maintained threads
        /// </summary>
        private readonly LinkedList<Worker> workers;

        /// <summary>
        /// The main lock root
        /// </summary>
        private readonly object mainLock = new object();

        /// <summary>
        /// Lock object for termination event
        /// </summary>
        //private readonly object terminationLock = new object();

        private readonly ManualResetEvent terminationLock = new ManualResetEvent(false);

        /// <summary>
        /// Current maintaning thread counts
        /// </summary> 
        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
        //it won't actually work in some case. Please refer
        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
        private volatile int currentThreadCount;
        //private volatile uint currentThreadCount;

        /// <summary>
        /// Current maintaning busy thread counts
        /// </summary>
        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
        //it won't actually work in some case. Please refer
        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
        private volatile int currentBusyThreadCount;
        //private volatile uint currentBusyThreadCount;

        /// <summary>
        /// Blocking <see cref="Queue"/> call or not
        /// </summary>
        private volatile bool blockQueueWhenFull;

        /// <summary>
        /// Current running state (running, shutting down, stop, terminated)
        /// </summary>
        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
        //it won't actually work in some case. Please refer
        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
        private volatile int runState;
        //private int runState;

        /// <summary>
        /// Reject handler, can be null
        /// </summary>
        private volatile ExecutorRejectHandler<T> rejectHandler;

        /// <summary>
        /// Exception handler, can be null
        /// </summary>
        private volatile ExecutorExceptionHandler<T> exceptionHandler;

        /// <summary>
        /// Gets or Set bare number of threads will be maintained after idle
        /// </summary>
        public int MinThread
        {
            get
            {
                return this.minThead;
            }

            private set
            {
                if (value > 0)
                {
                    this.minThead = value;
                }
            }
        }

        /// <summary>
        /// Gets or set maximum allowed thread
        /// </summary>
        public int MaxThread
        {
            get
            {
                return this.maxThead;
            }

            private set
            {
                if (value > 0)
                {
                    this.maxThead = value;
                    if (this.maxThead < this.MinThread)
                    {
                        this.MinThread = this.maxThead;
                    }
                }
            }
        }

        /// <summary>
        /// Gets or sets number of milliseconds that to maintain a thread if it fall to idle state
        /// Minimum accept value is 200 ms
        /// </summary>
        public int KeepAliveTime
        {
            get
            {
                return this.keepAliveTime;
            }

            set
            {
                if (value >= 200)
                {
                    this.keepAliveTime = value;
                }
            }
        }

        /// <summary>
        /// Gets or sets Block queue when full or not
        /// </summary>
        public bool BlockQueueWhenFull
        {
            get
            {
                return this.blockQueueWhenFull;
            }

            private set
            {
                this.blockQueueWhenFull = value;
            }
        }

        /// <summary>
        /// Gets or sets the reject handler
        /// </summary>
        public ExecutorRejectHandler<T> RejectHandler
        {
            get
            {
                return this.rejectHandler;
            }

            set
            {
                this.rejectHandler = value;
            }
        }

        /// <summary>
        /// Gets or sets the exception handler
        /// </summary>
        public ExecutorExceptionHandler<T> ExceptionHandler
        {
            get
            {
                return this.exceptionHandler;
            }

            set
            {
                this.exceptionHandler = value;
            }
        }

        /// <summary>
        /// Gets or sets name
        /// </summary>
        public string Name { get; set; }

        /// <summary>
        /// Initializes a new instance of the <see cref="JobExecutorEx{T}"/> class. 
        /// With MinThread = 5, MaxThread = 10, KeepAliveTime = 5000, MaximumQueueSize = 1000
        /// </summary>
        public NonQueueJobExcutor()
            : this(1, 5000)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="JobExecutorEx{T}"/> class. 
        /// </summary>
        /// <param name="thread">
        /// Number of minimum threads
        /// </param>
        /// <param name="keepAliveTime">
        /// Number of milliseconds to maintain an idle thread
        /// </param>
        public NonQueueJobExcutor(
            int thread, int keepAliveTime)
        {
            this.MinThread = thread;
            this.MaxThread = thread;
            this.KeepAliveTime = keepAliveTime;
            this.BlockQueueWhenFull = false;
            this.jobQueue = new WaitBlockingCollection<T>(thread);
            this.workers = new LinkedList<Worker>();
        }

        /// <summary>
        /// Initialize
        /// </summary>
        public void Initialize()
        {
        }

        /// <summary>
        /// Ensure the queued job is handled if the pool is not running and there is no worker
        /// </summary>
        /// <param name="job">
        /// The job to be executed
        /// </param>
        private void EnsureQueuedJobHandled(T job)
        {
            var reject = false;
            Worker t = null;
            lock (this.mainLock)
            {
                var state = this.runState;
                if (state != RUNNING)
                {
                    T item;
                    if (this.jobQueue.TryRemove(out item) && (job == item))
                    {
                        reject = true;
                    }
                }
                else if (state < STOP && this.currentThreadCount < Math.Max(this.MaxThread, 1) &&
                         (!(this.jobQueue.Count > 0)))
                {
                    t = this.AddThread(null);
                }
            }

            if (reject)
            {
                this.Reject(job, ExecutorRejectReason.ShuttingDown);
            }
            else if (t != null)
            {
                // ExecuteLoop a null job
                t.Start();
            }
        }

        /// <summary>
        /// Reject a job with reason by call to reject handler
        /// </summary>
        /// <param name="job">
        /// The rejected job
        /// </param>
        /// <param name="reason">
        /// The rejected reason
        /// </param>
        private void Reject(T job, ExecutorRejectReason reason)
        {
            var handler = this.RejectHandler;
            if (handler != null)
            {
                handler.Invoke(this, job, reason);
            }
        }

        /// <summary>
        /// Reject a job with reason by call to reject handler
        /// </summary>
        /// <param name="job">
        /// The job cause exception
        /// </param>
        /// <param name="exception">
        /// The occured exception
        /// </param>
        private void NotifyException(T job, Exception exception)
        {
            var handler = this.exceptionHandler;
            if (handler != null)
            {
                handler.Invoke(this, job, exception);
            }
        }

        /// <summary>
        /// Add a new thread if current thread count is less than minimum count
        /// </summary>
        /// <param name="job">
        /// The job to be executed
        /// </param>
        /// <returns>
        /// true if ok to create and add thread
        /// </returns>
        private bool AddThreadIfUnderMin(T job)
        {
            Worker t = null;
            lock (this.mainLock)
            {
                if (this.currentThreadCount < this.MinThread && this.runState == RUNNING)
                {
                    t = this.AddThread(job);
                }
            }

            if (t == null)
            {
                return false;
            }

            t.Start();
            return true;
        }

        /// <summary>
        /// Add a new thread if current thread count is less than maximum count
        /// </summary>
        /// <param name="job">
        /// The job to be executed
        /// </param>
        /// <returns>
        /// true if ok to create and add thread
        /// </returns>
        private bool AddThreadIfUnderMax(T job)
        {
            Worker t = null;
            lock (this.mainLock)
            {
                if ((this.currentThreadCount < this.MaxThread) && (this.currentBusyThreadCount == this.currentThreadCount) && this.runState == RUNNING)
                {
                    t = this.AddThread(job);
                }
            }

            if (t == null)
            {
                return false;
            }

            t.Start();
            return true;
        }

        /// <summary>
        /// Add a new thread to thread list
        /// </summary>
        /// <param name="firstJob">
        /// The first Job.
        /// </param>
        /// <returns>
        /// The thread has been just created and added
        /// </returns>
        private Worker AddThread(T firstJob)
        {
            var thread = new Worker(this, firstJob);
            this.workers.AddLast(thread);
            //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
            //it won't actually work in some case. Please refer
            //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
            Interlocked.Increment(ref this.currentThreadCount);
            //this.currentThreadCount++;
            return thread;
        }

        /// <summary>
        /// Called by threads to get for the next job
        /// Will be block to wait for event from Queue method if there is no job in the jobqueue
        /// </summary>
        /// <returns>
        /// Next job from the queue
        /// </returns>
        private T GetJob()
        {
            while (true)
            {
                try
                {
                    var state = this.runState;
                    if (state > SHUTDOWN)
                    {
                        return null;
                    }

                    T r;
                    if (state == SHUTDOWN)
                    {
                        // Help drain queue
                        this.jobQueue.TryRemove(out r);
                    }
                    else if (this.currentThreadCount > this.MinThread)
                    {
                        // Only try to get new job in a specified time if timeout the free to exit
                        this.jobQueue.TryRemove(out r, this.keepAliveTime);
                    }
                    else
                    {
                        r = this.jobQueue.Remove();
                    }

                    if (r != null)
                    {
                        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                        //it won't actually work in some case. Please refer
                        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                        Interlocked.Increment(ref this.currentBusyThreadCount);
                        //currentBusyThreadCount++;
                        return r;
                    }

                    if (this.WorkerCanExit())
                    {
                        // Wake up others if is shutting down
                        if (this.runState >= SHUTDOWN)
                        {
                            this.InterruptIdleThreads();
                        }

                        return null;
                    }
                }
                catch (ThreadInterruptedException)
                {
                    // Expected exception
                    // On interruption, re-check runState
                }
            }
        }

        /// <summary>
        /// Interup idle threads
        /// </summary>
        private void InterruptIdleThreads()
        {
            lock (this.mainLock)
            {
                foreach (var t in this.workers)
                {
                    t.InteruptIfIdle();
                }
            }
        }

        /// <summary>
        /// Check whether a worker thread that fails to get a task can
        /// exit.  We allow a worker thread to die if the pool is stopping,
        /// or the queue is empty
        /// </summary>
        /// <returns>
        /// True if the thread can exit
        /// </returns>
        private bool WorkerCanExit()
        {
            bool canExit;
            lock (this.mainLock)
            {
                canExit = this.runState >= STOP || this.jobQueue.Count == 0;
            }

            return canExit;
        }

        /// <summary>
        /// Will be perform when a thread exit
        /// </summary>
        /// <param name="worker">
        /// The worker.
        /// </param>
        private void WorkDone(Worker worker)
        {
            lock (this.mainLock)
            {
                this.workers.Remove(worker);
                //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                //it won't actually work in some case. Please refer
                //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                Interlocked.Decrement(ref this.currentThreadCount);
                //this.currentThreadCount--;
                this.TryTerminate();
            }
        }

        /// <summary>
        /// Try to terminate the executor
        /// </summary>
        private void TryTerminate()
        {
            if (this.currentThreadCount == 0)
            {
                var state = this.runState;

                // Still have job to do
                if (state < STOP && this.jobQueue.Count > 0)
                {
                    // disable termination check below
                    state = RUNNING;
                    var t = this.AddThread(null);
                    if (t != null)
                    {
                        t.Start();
                    }
                }

                if (state == STOP || state == SHUTDOWN)
                {
                    //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                    //it won't actually work in some case. Please refer
                    //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                    Interlocked.Exchange(ref this.runState, TERMINATED);
                    //this.runState = TERMINATED;
                    this.terminationLock.Set();
                    //lock (this.terminationLock)
                    //{
                    //    Monitor.PulseAll(this.terminationLock);
                    //}
                }
            }
        }

        /// <summary>
        /// Drain the job queue
        /// </summary>
        /// <returns>
        /// List of remain jobs
        /// </returns>
        private List<T> DrainQueue()
        {
            var taskList = new List<T>();
            while (this.jobQueue.Count > 0)
            {
                var consumingEnum = this.jobQueue.GetConsumingEnumerable();
                foreach (var job in consumingEnum)
                {
                    taskList.Add(job);
                }
            }

            return taskList;
        }

        /// <summary>
        /// Wait for the executor terminated
        /// If timeout is zero, it will wait forever.
        /// </summary>
        /// <param name="timeout">
        /// Timeout icn milliseconds
        /// </param>
        /// <returns>
        /// True if the executor ternminated
        /// </returns>
        public bool WaitForTermination(int timeout)
        {
            //lock (this.terminationLock)
            //{
                if (this.runState == TERMINATED)
                {
                    return true;
                }

                try
                {
                    // Support wait forever when timeout is zero
                    if (timeout == 0)
                    {
                        timeout = System.Threading.Timeout.Infinite;
                    }

                    terminationLock.WaitOne(timeout);
                    terminationLock.Reset();
                    //Monitor.Wait(this.terminationLock, timeout);
                }
                catch (ThreadAbortException)
                {
                }
                catch (ThreadStateException)
                {
                }
                catch (ThreadInterruptedException)
                {
                }

                if (this.runState != TERMINATED)
                {
                    return false;
                }

                return true;
            //}
        }

        /// <summary>
        /// Dispose the object
        /// </summary>
        public void Dispose()
        {
            this.Stop(true);
        }

        /// <summary>
        /// Enqueue a job
        /// Increase thread count if nessesary
        /// Also careful handle for shutting down case, reject job if shutting down
        /// </summary>
        /// <param name="job">
        /// The job should be done
        /// </param>
        public void Queue(T job)
        {
            if (job == null)
            {
                throw new NullReferenceException("Job should not null");
            }

            // Fill min thread
            if (this.currentThreadCount < this.MinThread && this.AddThreadIfUnderMin(job))
            {
                //Logger.WarnFormat("{0}: Job is rejected since Current thread count [{1}] less than Min thread count [{2}]",
                //    this.Name, this.currentThreadCount, this.MinThread);
                return;                
            }

            // All thread is busy and we have not reached the max thread count yet
            if (this.AddThreadIfUnderMax(job))
            {
                return;
            }

            // check has free thread
            if (this.currentBusyThreadCount == this.MaxThread)
            {
                return;
            }

            // Try add to queue
            if (this.runState == RUNNING && this.jobQueue.TryAdd(job))
            {
                if (this.runState != RUNNING || this.currentThreadCount == 0)
                {
                    this.EnsureQueuedJobHandled(job);
                }

                return;
            }

            // Add failed try to reject or block
            if (this.runState == RUNNING)
            {
                if (this.blockQueueWhenFull)
                {
                    // Do not worry about waiting forever as when shutting down the queue will be drained
                    this.jobQueue.Add(job);
                    if (this.runState != RUNNING || this.currentThreadCount == 0)
                    {
                        this.EnsureQueuedJobHandled(job);
                    }
                }
                else
                {
                    this.Reject(job, ExecutorRejectReason.TooBusy);
                    Logger.WarnFormat("{0}: Job is rejected since The executor is too busy for the new request", this.Name);
                }
            }
            else
            {
                this.Reject(job, ExecutorRejectReason.ShuttingDown);
                Logger.WarnFormat("{0}: Job is rejected since The executor is shutting down", this.Name);
            }
        }

        /// <summary>
        /// Try to stop the executor by prevent new message queuing and notify all worker threads
        /// First change the run state to stop then interup any idle worker thread waiting for job,
        /// if there is no threads or tasks then terminate by tryTerminate if not then the last worker will
        /// need to turn off the lights
        /// </summary>
        /// <param name="force">
        /// Force stop by finish all executing thread and 
        /// </param>
        /// <returns>
        /// List of remain jobs if force
        /// </returns>
        public List<T> Stop(bool force)
        {
            lock (this.mainLock)
            {
                List<T> remainJobs;
                var state = this.runState;
                if (force)
                {
                    if (state < STOP)
                    {
                        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                        //it won't actually work in some case. Please refer
                        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                        Interlocked.Exchange(ref this.runState, STOP);
                        //this.runState = STOP;
                    }

                    foreach (var t in this.workers)
                    {
                        t.Interupt();
                    }

                    remainJobs = new List<T>();//this.DrainQueue();
                }
                else
                {
                    if (state < SHUTDOWN)
                    {
                        //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                        //it won't actually work in some case. Please refer
                        //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                        Interlocked.Exchange(ref this.runState, SHUTDOWN);
                        //this.runState = SHUTDOWN;
                    }

                    this.InterruptIdleThreads();

                    // foreach (var t in this.workers)
                    // {
                    // t.InteruptIfIdle();
                    // }
                    remainJobs = new List<T>();
                }

                // Terminate now if pool and queue empty
                this.TryTerminate();
                return remainJobs;
            }
        }

        /// <summary>
        /// Worker thread
        /// </summary>
        private class Worker
        {
            /// <summary>
            /// The job executor
            /// </summary>
            private readonly NonQueueJobExcutor<T> executor;

            /// <summary>
            /// The first job to execute
            /// </summary>
            private T firstJob;

            /// <summary>
            /// The managed thread to run
            /// </summary>
            private readonly Thread thread;

            /// <summary>
            /// The worker lock object
            /// </summary>
            private readonly object lockObj;

            /// <summary>
            /// Initializes a new instance of the <see cref="Worker"/> class. 
            /// worker.
            /// </summary>
            /// <param name="executor">
            /// The executor.
            /// </param>
            /// <param name="firstJob">
            /// The first Job
            /// </param>
            public Worker(NonQueueJobExcutor<T> executor, T firstJob)
            {
                this.executor = executor;
                if (firstJob != null)
                {
                    //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                    //it won't actually work in some case. Please refer
                    //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                    Interlocked.Increment(ref this.executor.currentBusyThreadCount);
                    //this.executor.currentBusyThreadCount++;
                }
                
                this.firstJob = firstJob;
                this.lockObj = new object();
                this.thread = new Thread(this.ExecuteLoop);
                if (executor.Name != null)
                {
                    this.thread.Name = executor.Name + "_" + this.thread.ManagedThreadId;
                }
            }

            /// <summary>
            /// Start the worker
            /// </summary>
            public void Start()
            {
                thread.Start();
            }

            /// <summary>
            /// Execute loop for a thread
            /// </summary>
            private void ExecuteLoop()
            {
                try
                {
                    var job = firstJob;
                    firstJob = null;
                    while ((job != null) || (job = executor.GetJob()) != null)
                    {
                        if (this.EnterLock())
                        {                            
                            try
                            {
                                try
                                {
                                    job.Execute();
                                }
                                catch (Exception ex)
                                {
                                    executor.NotifyException(job, ex);
                                }
                            }
                            finally
                            {
                                Monitor.Exit(lockObj);
                            }
                            //Tuan Nguyen - Fix bug: SPR 116187: don't use volatile because
                            //it won't actually work in some case. Please refer
                            //http://stackoverflow.com/questions/154551/volatile-vs-interlocked-vs-lock
                            Interlocked.Decrement(ref this.executor.currentBusyThreadCount);
                            //executor.currentBusyThreadCount--;
                            job = null;

                            // 1 more job processed by JobExecutorEx
                            var count = Interlocked.Increment(ref executor.intervalQueueSizeCheck);
                            if (count >= 2000)
                            {
                                lock (executor)
                                {
                                    //Logger.InfoFormat("{0}: JobExecutorEx Queue Size: {1}", executor.Name, executor.jobQueue.Count);
                                    executor.intervalQueueSizeCheck = 0;
                                }
                            }
                        }
                        else
                        {
                            break;
                        }
                    }
                }
                finally
                {
                    executor.WorkDone(this);
                }
            }

            /// <summary>
            /// Check interupt is called on current thread and enter lock
            /// </summary>
            /// <returns>
            /// true if can enter lock
            /// </returns>
            private bool EnterLock()
            {
                try
                {
                    Monitor.Enter(lockObj);
                    return true;
                }
                catch (ThreadInterruptedException)
                {
                    if (executor.runState >= STOP)
                    {
                        return false;
                    }

                    Monitor.Enter(lockObj);
                    return true;
                }
            }

            /// <summary>
            /// Safe interupt the worker thread
            /// </summary>
            public void Interupt()
            {
                try
                {
                    thread.Interrupt();
                }
                catch (ThreadAbortException)
                {
                }
                catch (ThreadStateException)
                {
                }
                catch (ThreadInterruptedException)
                {
                }
            }

            /// <summary>
            /// Interupt the worker thread if waiting for job
            /// </summary>
            public void InteruptIfIdle()
            {
                if (this.thread != Thread.CurrentThread)
                {
                    if (Monitor.TryEnter(lockObj))
                    {
                        try
                        {
                            this.thread.Interrupt();
                        }
                        finally
                        {
                            Monitor.Exit(lockObj);
                        }
                    }
                }
            }
        }
    }
}